package cn.dmp.report

import cn.dmp.beans.Log
import cn.dmp.report.AppDictAnalyse.args
import cn.dmp.utils.{JedisPools, RptUItils}
import org.apache.commons.lang.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SaveMode, SparkSession}
import redis.clients.jedis.{Jedis, JedisPool}


/**
  * 需求：从redis数据库里读取字典文件数据，然后和日志文件数据匹配。
  *
  */
object AppDictAnalyse_Version2 {
  def main(args: Array[String]): Unit = {
    // 0.判断参数
    if (args.length != 3){
      println(
        """
          |cn.dmp.report.ProCityRpt_Version3
          |参数：
          |   InputPath
          |   dictFilePath
          |   resultOutputPath
        """.stripMargin)
    }

    // 1.接收程序参数
    val Array(inputPath, dictFilePath, resultOutputPath) = args

    // 2.创建SparkSession
    val sparkSession: SparkSession = SparkSession.builder().appName("ProCityRpt_Version3").master("local[*]")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //spark优化：spark.serializer修改磁盘序列化方式
      .enableHiveSupport()
      .getOrCreate()

    /**
      * 读取日志数据 和 redis数据。进行匹配
      *
      */
    val dataSet: Dataset[String] = sparkSession.read.textFile(inputPath)

    val res: Dataset[(String, List[Double])] = dataSet.map(line => line.split(",", -1)).filter(_.length >= 85)
      .map(Log(_)).filter(log => !log.Appid.isEmpty || !log.Appname.isEmpty)
      .mapPartitions(itr => {
        //有数据库连接的话最好是分个mapPartitions，在其里面连接然后在map
        val jedis: Jedis = JedisPools.getJedis()

        val parResult = new collection.mutable.ListBuffer[(String, List[Double])]()

        //遍历分区的所有数据，查询redis，目的是把Appname为空的数据进行转换，将结果放入到ListBuffer中
        itr.foreach(log => {
          var appname: String = log.Appname
          if (StringUtils.isEmpty(appname)) {
            appname = jedis.get(log.Appid)
          }

          val req: List[Double] = RptUItils.caculateReq(log.Requestmode, log.Processnode)
          val rtb: List[Double] = RptUItils.caculateRtb(log.Iseffective, log.Isbilling, log.Isbid, log.Adorderid, log.Iswin, log.Winprice, log.Adpayment)
          val showClick: List[Double] = RptUItils.caculateShowClick(log.Requestmode, log.Iseffective)

          parResult += ((appname, req ++ rtb ++ showClick)) //foreach与map的区别：前者无返回数据，后者有返回数据，
        })
        jedis.close()
        parResult.iterator //因为mapPartitions要求返回的是迭代器，ListBuffer数据可以转换成迭代器！

        /**
          * 如果前面用的是map，map是有返回值，默认返回最后一行代码，
          * 但是这里又不能返回jedis.close。因此还是用foreach，但是foreach没有返回值，因此需要定义一个ListBuffer来保存数据，来返回一个迭代器
          *
          */
      })
    res.write.mode(SaveMode.Overwrite).text(resultOutputPath)

    //本地测试运行：!!!!!注意提前设置好参数：输入目录和输出目录
    /**
      * 输入目录：E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\data\2016-10-01_06_p1_invalid.1475274123982.log.FINISH
      * 输入目录：字典文件数据
      * 输出目录：E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\AppDictAnalyse_Version2
      */


    sparkSession.stop()


  }
}
